#pragma once
#include "../common/message.hpp"
#include "../common/net.hpp"
#include <unordered_set>

namespace qiangsir
{
    namespace server
    {
        class TopicManage
        {

        public:
            using ptr = std::shared_ptr<TopicManage>;
            void onTopicService(const BaseConnection::ptr &conn, TopicRequest::ptr &msg)
            {
                TopicOptype topicoptype = msg->optype();
                bool ret = true;
                switch (topicoptype)
                {
                    // 1.主题创建
                    // 2.主题订阅
                    // 3.主题删除
                    // 4.主题取消订阅
                    // 5.主题消息发布
                case TopicOptype::TOPIC_CREATE:
                    topicCreate(conn, msg);
                    break;
                case TopicOptype::TOPIC_REMOVE:
                    topicRemove(conn, msg);
                    break;
                case TopicOptype::TOPIC_SUBSCRIBE:
                    ret = topicSubseribe(conn, msg);
                    break;
                case TopicOptype::TOPIC_CANCEL:
                    topicCancel(conn, msg);
                    break;
                case TopicOptype::TOPIC_PUBLISH:
                    ret = topicPublish(conn, msg);
                    break;
                default:
                    return errresponse(conn, msg, RCode::RCODE_INVALID_OPTYPE);
                }
                if (!ret)
                    return errresponse(conn, msg, RCode::RCODE_NOT_FOUND_TOPIC);
                    return topicResponse(conn,msg);
            }
            void shutDownConn(const BaseConnection::ptr &conn)
            {
                std::vector<Topic::ptr> topics;
                Subsriber::ptr subsriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _subsribers.find(conn);
                    if(it == _subsribers.end())
                    {
                        return;
                    }
                    subsriber = it->second;
                    for(auto topic:subsriber->topics)
                    {
                        auto topic_it = _topics.find(topic);
                        if(topic_it == _topics.end()) 
                        continue;
                        topics.push_back(topic_it->second);
                    }
                    _subsribers.erase(it);
                }
                for(auto &topic:topics)
                {
                    topic->delSubsriber(subsriber);
                }
            }

        private:
            void errresponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg, RCode rcode)
            {
                auto resp_msg = BaseMessageFactory::create<TopicResponse>();
                resp_msg->setId(msg->id());
                resp_msg->setType(MType::RSP_TOPIC);
                resp_msg->setRcode(rcode);
                conn->send(resp_msg);
            }
            void topicResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                auto msg_rsp = BaseMessageFactory::create<TopicResponse>();
                msg_rsp->setId(msg->id());
                msg_rsp->setType(MType::RSP_TOPIC);
                msg_rsp->setRcode(RCode::RCODE_OK);
                conn->send(msg_rsp);
            }
            void topicCreate(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                std::string topic_name = msg->topicKey();
                Topic::ptr topic = std::make_shared<Topic>(topic_name);
                _topics.insert(std::make_pair(topic_name, topic));
            }
            void topicRemove(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 1. 查看当前主题，有哪些订阅者，然后从订阅者中将主题信息删除掉
                // 2. 删除主题的数据 -- 主题名称与主题对象的映射关系
                std::string topic_name = msg->topicKey();
                std::unordered_set<Subsriber::ptr> subsribers;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _topics.find(topic_name);
                    if (it == _topics.end())
                    {
                        return;
                    }
                    subsribers = it->second->subsribers;
                    _topics.erase(it);
                }
                // 找出订阅者一个个切断订阅
                for (auto subsribe : subsribers)
                {
                    subsribe->delTopic(topic_name);
                }
            }
            bool topicSubseribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 寻找主题和订阅者
                // 找不到主题需要报错，找不到订阅者需要创建订阅者
                std::string topic_name = msg->topicKey();
                Topic::ptr topic;
                Subsriber::ptr subsriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _topics.find(topic_name);
                    if (it == _topics.end())
                    {
                        return false;
                    }
                    topic = it->second;
                    auto sub_it = _subsribers.find(conn);
                    if (sub_it != _subsribers.end())
                    {
                        subsriber = sub_it->second;
                    }
                    else
                    {
                        subsriber = std::make_shared<Subsriber>(conn);
                        _subsribers.insert(std::make_pair(conn, subsriber));
                    }
                }
                // 开始订阅
                subsriber->appendTopic(topic_name);
                topic->appendSubsriber(subsriber);
                return true;
            }
            void topicCancel(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 主题不存在报错，订阅者不存在直接返回
                std::string topic_name = msg->topicKey();
                Topic::ptr topic;
                Subsriber::ptr subsriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _topics.find(topic_name);
                    if (it != _topics.end())
                    {
                        topic = it->second;
                    }
                    auto sub_it = _subsribers.find(conn);
                    if (sub_it != _subsribers.end())
                    {
                        subsriber = sub_it->second;
                    }
                }
                if (topic)
                    topic->delSubsriber(subsriber);
                if (subsriber)
                    subsriber->delTopic(topic_name);
            }
            bool topicPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                std::string topic_name = msg->topicKey();
                Topic::ptr topic;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _topics.find(topic_name);
                    if (it == _topics.end())
                    {
                        return false;
                    }
                    topic = it->second;
                }
                topic->pushMessage(msg);
                return true;
            }

        private:
            struct Subsriber
            {
                using ptr = std::shared_ptr<Subsriber>;
                std::mutex mutex;
                BaseConnection::ptr conn;
                std::unordered_set<std::string> topics;

                Subsriber(const BaseConnection::ptr &c) : conn(c) {}
                void appendTopic(const std::string &name)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    topics.insert(name);
                }
                void delTopic(const std::string &name)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    topics.erase(name);
                }
            };
            struct Topic
            {
                using ptr = std::shared_ptr<Topic>;
                std::mutex mutex;
                std::string name;
                std::unordered_set<Subsriber::ptr> subsribers;

                Topic(const std::string &topic_name) : name(topic_name) {}
                void appendSubsriber(const Subsriber::ptr &subsriber)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    subsribers.insert(subsriber);
                }
                void delSubsriber(const Subsriber::ptr &subsriber)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    subsribers.erase(subsriber);
                }
                void pushMessage(const BaseMessage::ptr &msg)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    for (auto subsriber : subsribers)
                    {
                        subsriber->conn->send(msg);
                    }
                }
            };

        private:
            std::mutex _mutex;
            std::unordered_map<std::string, Topic::ptr> _topics;
            std::unordered_map<BaseConnection::ptr, Subsriber::ptr> _subsribers;
        };
    }
}